package com.bkjk.mongoToMysql;

import java.io.File;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.Queue;
import java.util.Set;

import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;

import org.pentaho.di.core.plugins.PluginRegistry;
import org.pentaho.di.core.plugins.StepPluginType;
import org.pentaho.di.trans.TransHopMeta;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.steps.datagrid.DataGridMeta;
import org.pentaho.di.trans.steps.joinrows.JoinRowsMeta;
import org.pentaho.di.trans.steps.mongodbinput.MongoDbInputMeta;
import org.pentaho.di.trans.steps.tableoutput.TableOutputMeta;
import org.pentaho.mongo.wrapper.field.MongoField;

import com.bkjk.constant.DataBaseMetaConstant;
import com.bkjk.constant.KettleConstant;
import com.bkjk.dbConn.MongoDao;
import com.bkjk.utils.KettleUtils;

/**
 * 生成非自增表的ktr
 * 
 * 转换问题：mysql中数据是大小写不敏感的 解决方案:将mysql设置为大小写敏感的 建库时排序规则选择utf_bin
 * 
 * 对于每一个在mongo中自动增加的表，都映射为同一张表
 * 如app_crashesA表和app_crashesB表都将数据插入到mysql中的app_crashes表
 * 
 * 
 * 该类用来对自动增加的表进行处理
 * @author liangjie
 *
 */
public class GeneIncreKtr

{
	private final boolean isTest = true;
	private MongoDao mongoDao = MongoDao.getInstance();;

	public GeneIncreKtr() {
		// 初次执行，需要初始化kettle相关的env
		try {
			KettleUtils.initEnv();
			/*
			 * File file = new File(KettleConstant.INCRE_FILE_DIR);
			 * if(!file.exists()){ file.mkdir(); }
			 */
		} catch (KettleException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	public void transForEveryCollection() {
		//初始化获取countly中自增集合和非自增集合
		mongoDao.init_NonOrIncreCollection("countly");
		// 获取对象,这些collectionNames包含了所有的名称
		Queue<String> collections = mongoDao.getIncreaseCollections();
		// 针对每一个自动增加的集合
		for (final String collectionName : collections) {
			//获取集合对应的mysql中的表名称
			final String mysqlTableName = KettleUtils.mapIncreCollectionToTableName(collectionName);
			//调用创建转换的方法
			TransMeta transMeta = createTransMeta(collectionName, mysqlTableName);
			Optional<TransMeta> result = Optional.ofNullable(transMeta);
			//使用lambda表达式处理非空数据
			result.ifPresent(value -> {
				try {
					writeToXML(value, collectionName, mysqlTableName);
					KettleUtils.executeTrans(transMeta);
					// 睡眠一秒
					Thread.sleep(1000);
				} catch (Exception e) {
					// TODO Auto-generated catch block
					e.printStackTrace();

					// err.put(collectionName, e.getMessage());
				}
			});

		}
	}

	public TransMeta createTransMeta(String collectionName, String mysqlTableName) {

		// 针对每一个collectionName+appId创建一个转换
		// Trans trans = KettleUtils.getTrans();
		// 针对每一个转换，创建一个转换meta

		TransMeta transMeta = KettleUtils.getTransMeta();

		// 获取集合中的所有列
		Set<String> columns = mongoDao.getColumnsInCollection(collectionName, "countly");
		// 如果队列为空，那么创建一个空的输出
		if (columns.size() == 0) {
			return null;
		}
		//
		PluginRegistry registry = PluginRegistry.getInstance();
		// 针对每一个过来的collection，需要增加appID

		// 创建文件夹
		createIncreDir(mysqlTableName);

		transMeta.setName(collectionName);
		// 设置trans的transMeta
		// trans.setTransMeta(transMeta);

		// 针对每一个转换meta，创建一个Mongo输入
		MongoDbInputMeta mongoDbInputMeta = CustomPlugin.getCustomMongoInput();
		// 设置集合的名称

		mongoDbInputMeta.setCollection(collectionName);

		List<MongoField> mongoFields = new ArrayList<>();
		// 用来设置表输出,因为自增表需要增加一个tableId以区分不同表的数据,所以输出表中的列需要增加一列
		int outputFieldSize = columns.size() + 1;

		String[] fieldDatabases = new String[outputFieldSize];
		// 用来设置表输出
		String[] fieldStream = new String[outputFieldSize];
		int index = 0;
		MongoField mongField = null;
		for (String column : columns) {
			mongField = new MongoField();
			mongField.m_fieldName = column;
			mongField.m_fieldPath = column;
			mongField.m_kettleType = "String";
			// mongField.setIndexedVals(column);
			mongoFields.add(mongField);
			// 将对应的列设置到fieldDatabases和fieldStream中
			fieldDatabases[index] = KettleUtils.mapColumnsToNonKey(column);
			fieldStream[index] = column;
			index++;

		}
		// 设置mongo不以单独的json输出方式输出
		mongoDbInputMeta.setOutputJson(false);

		mongoDbInputMeta.setMongoFields(mongoFields);

		// 设置mongo的输出
		// mongoDbInputMeta.setMongoFields();
		// List<MongoField> mongoFields = mongoDbInputMeta.getMongoFields();
		// mongoDbInputMeta.
		// mongoFields.stream().forEach(field->System.out.println(field.""));
		// 创建特定的stepMeta，已经设置了mongoInput的stepID
		StepMeta from_Mongo_stepMeta = CustomPlugin.getCustomMongoStepMeta();
		from_Mongo_stepMeta.setName("mongo集合[" + collectionName + "]");
		from_Mongo_stepMeta.setStepMetaInterface(mongoDbInputMeta);
		from_Mongo_stepMeta.setLocation(400, 200);
		// 将该步加入到转换中
		transMeta.addStep(from_Mongo_stepMeta);

		// 创建一个自定义常量数据
		// 如果不知道类名，从kettle的工具中，画一个该类型的组件，然后打开xml
		// 查看相应的type
		DataGridMeta dataGridMeta = KettleUtils.getDataGridMeta();

		String defaultValue = mysqlTableName;

		if (collectionName.startsWith(mysqlTableName)) {
			defaultValue = collectionName.substring(mysqlTableName.length());
		}
		dataGridMeta.setFieldName(new String[] { DataBaseMetaConstant.TABLEID });
		List<List<String>> datalines = new ArrayList<>();
		List<String> lis0 = new ArrayList<>();
		lis0.add(defaultValue);
		datalines.add(lis0);
		dataGridMeta.setDataLines(datalines);
		dataGridMeta.setFieldType(new String[] { "String" });

		// dgm的组件ID
		String dataGridStepMetaId = registry.getPluginId(StepPluginType.class, dataGridMeta);
		// 创建一个step来封装常量组件
		StepMeta dataGridStepMeta = KettleUtils.getStepMeta();
		dataGridStepMeta.setStepID(dataGridStepMetaId);
		dataGridStepMeta.setLocation(400, 600);
		dataGridStepMeta.setStepMetaInterface(dataGridMeta);
		dataGridStepMeta.setName("自定义常量数据");
		// 将该步加入到转换中
		transMeta.addStep(dataGridStepMeta);

		// 创建一个表连接
		// 根据测试，什么都不用设置，即可连接常量数据到表后
		JoinRowsMeta joinRowsMeta = KettleUtils.getJoinRowsMeta();

		String joinRowsMetaId = registry.getPluginId(StepPluginType.class, joinRowsMeta);
		StepMeta joinRowsStepMeta = KettleUtils.getStepMeta();
		joinRowsStepMeta.setStepID(joinRowsMetaId);
		joinRowsStepMeta.setStepMetaInterface(joinRowsMeta);
		joinRowsStepMeta.setName("关联俩个表");
		joinRowsStepMeta.setLocation(800, 400);
		// 将步骤添加到转换中
		transMeta.addStep(joinRowsStepMeta);

		// 创建一个表输出
		DatabaseMeta dbm = KettleUtils.getDataBase();
		dbm.setName("mysql测试");
		// 将db加入到trans中
		transMeta.addDatabase(dbm);
		// dbm.setObjectId(id);

		TableOutputMeta tableOutput = new TableOutputMeta();
		// 设置是否裁剪表
		// 由于所有表放在一个表中，不再设置trunc
		// 不知是否会因为主键导致导入失败
		//测试时，设置表裁剪
		if(isTest){
			tableOutput.setTruncateTable(true);
		}
		tableOutput.setDatabaseMeta(dbm);

		tableOutput.setTableName(mysqlTableName);

		// 设置指定数据库字段
		tableOutput.setSpecifyFields(true);
		// 获取mongo的collection的列，并将输入的列映射到其它的列
		fieldDatabases[outputFieldSize - 1] = DataBaseMetaConstant.TABLEID;
		fieldStream[outputFieldSize - 1] = DataBaseMetaConstant.TABLEID;
		tableOutput.setFieldDatabase(fieldDatabases);
		tableOutput.setFieldStream(fieldStream);

		String tostepid = registry.getPluginId(StepPluginType.class, tableOutput);
		// 新建一个步骤
		StepMeta to_stepMeta = KettleUtils.getStepMeta();
		to_stepMeta.setLocation(1200, 400);
		to_stepMeta.setStepMetaInterface(tableOutput);
		to_stepMeta.setName("["+collectionName+"]"+"到mysql[" + mysqlTableName + "]的输出");
		to_stepMeta.setStepID(tostepid);
		// 将步骤添加
		transMeta.addStep(to_stepMeta);

		// 建立相应的连接
		TransHopMeta mongoJoin = new TransHopMeta(from_Mongo_stepMeta, joinRowsStepMeta);
		TransHopMeta dataJoin = new TransHopMeta(dataGridStepMeta, joinRowsStepMeta);
		TransHopMeta joinToMysql = new TransHopMeta(joinRowsStepMeta, to_stepMeta);
		// 将相应的连接添加到表中
		transMeta.addTransHop(mongoJoin);
		transMeta.addTransHop(dataJoin);
		transMeta.addTransHop(joinToMysql);
		// 创建一个mysql的表输出

		return transMeta;

	}

	public void writeToXML(TransMeta transMeta, String collectionName, String mysqlTableName)
			throws UnsupportedEncodingException, IOException, KettleException {
		String fileName = KettleConstant.INCRE_FILE_DIR + mysqlTableName + File.separator + collectionName + "转换"
				+ KettleConstant.FILE_SUFFIX;
		KettleUtils.writeToXML(transMeta, fileName);
	}

	/**
	 * 针对每个不同的集合创建不同的文件夹
	 * 
	 * @param collectionName
	 */
	public void createIncreDir(String collectionName) {
		// 创建集合的文件
		File file = new File(KettleConstant.INCRE_FILE_DIR + collectionName);
		if (!file.exists()) {
			file.mkdir();
		}
	}

	public static void main(String[] args) {
		System.out.println("测试开始");
		try {
			KettleUtils.initEnv();
			String collectionName = "cities";
			String fileName = KettleConstant.FILE_DIR + collectionName + "转换" + KettleConstant.FILE_SUFFIX;
			GeneIncreKtr test = new GeneIncreKtr();
			/*
			 * List<TransMeta> transMetas =
			 * test.createTransMeta(collectionName); for(TransMeta
			 * transMeta:transMetas){ KettleUtils.writeToXML(transMeta,
			 * fileName); System.out.println("文件路径为:"+fileName); }
			 */
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		System.out.println("测试结束");
	}

	public void close() {
		if (mongoDao != null) {
			mongoDao.close();
		}
	}
}
